package com.fly.demo.web.sse;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

/**
 * 解决多客户端数据同时更新<br>
 * 
 * https://blog.csdn.net/demon7552003/article/details/111097553
 * 
 */
@Slf4j
@RestController
public class SSE2Controller
{
    List<Consumer<String>> producers = new CopyOnWriteArrayList<>();
    
    @GetMapping("/sse/v2")
    public Flux<ServerSentEvent<String>> subscribe()
    {
        // 为每个用户创建一个独立的SSE流
        return Flux.create(sink -> {
            Consumer<String> producer = nextData -> sink.next(nextData);
            producers.add(producer);
            sink.onDispose(() -> producers.remove(producer)); // 用户断开连接时移除
        }).map(data -> ServerSentEvent.<String> builder().data(String.valueOf(data)).build());
    }
    
    /**
     * 支持多客户端的sse数据更新
     */
    @Scheduled(cron = "0/10 * 7-23 * * ?")
    private void publish()
    {
        log.info("##### {}", producers.size());
        String message = DateFormatUtils.format(System.currentTimeMillis(), "yyyy-MM-dd HH:mm:ss.SSS");
        producers.stream().parallel().forEach(p -> p.accept(message));
    }
}